package cn.chrilwe.pigpig.im.rabbit.factory.impl;

import java.io.IOException;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.TimeoutException;

import org.springframework.context.ApplicationContext;

import com.alibaba.fastjson.JSON;
import com.rabbitmq.client.AMQP.Basic.Consume;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

import cn.chrilwe.pigpig.im.model.RabbitModel;
import cn.chrilwe.pigpig.im.protocol.TalkRequest;
import cn.chrilwe.pigpig.im.protocol.TalkResponse;
import cn.chrilwe.pigpig.im.rabbit.factory.RabbitSession;
import cn.chrilwe.pigpig.im.server.WebSocketServer;
import cn.chrilwe.pigpig.im.server.handler.MessageHandler;
import cn.chrilwe.pigpig.im.service.MessageService;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import lombok.extern.slf4j.Slf4j;

/**
 * 
 * @author chrilwe 2020-8-16
 */
@Slf4j
public class RabbitSessionTemplate implements RabbitSession {

	public Connection connection;

	public RabbitSessionTemplate(Connection connection) {
		this.connection = connection;
	}

	@Override
	public Connection getConnection() {

		return connection;
	}

	@Override
	public void send(String message, RabbitModel rabbitModel) {
		Channel channel = null;
		try {
			channel = connection.createChannel();
			channel.exchangeDeclare(rabbitModel.getExchange(), BuiltinExchangeType.TOPIC);
			channel.queueDeclare(rabbitModel.getQueue(), rabbitModel.isPersistence(),
					rabbitModel.isCurrentConnectPrivate(), rabbitModel.isAutoDelete(), rabbitModel.getMap());
			channel.queueBind(rabbitModel.getQueue(), rabbitModel.getExchange(), rabbitModel.getRoutingKey());
			channel.basicPublish(rabbitModel.getExchange(), rabbitModel.getRoutingKey(), null, message.getBytes());
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} finally {
			try {
				channel.close();
				connection.close();
			} catch (IOException | TimeoutException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
	}

	@Override
	public void receive(RabbitModel rabbitModel) {
		try {
			Channel channel = connection.createChannel();
			channel.exchangeDeclare(rabbitModel.getExchange(), BuiltinExchangeType.TOPIC);
			channel.queueDeclare(rabbitModel.getQueue(), rabbitModel.isPersistence(),
					rabbitModel.isCurrentConnectPrivate(), rabbitModel.isAutoDelete(), rabbitModel.getMap());
			channel.queueBind(rabbitModel.getQueue(), rabbitModel.getExchange(), rabbitModel.getRoutingKey());
			DefaultConsumer consume = new DefaultConsumer(channel) {

				@Override
				public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
						byte[] body) throws IOException {
					log.debug("========接收到其他服务器的转发消息: " + new String(body) + "=======");
					TalkRequest request = JSON.parseObject(new String(body), TalkRequest.class);
					if (request != null) {
						// 消息处理
						ApplicationContext context = WebSocketServer.applicationContext;
						MessageService service = context.getBean(MessageService.class);
						TalkResponse res = service.topicMsg(request);

						Map<String, Map> groupChannel = MessageHandler.groupChannel;
						Map<String, io.netty.channel.Channel> map = groupChannel.get(request.getRoomId() + "");
						if(map != null && map.size() > 0) {
							Set<Entry<String, io.netty.channel.Channel>> entrySet = map.entrySet();
							for (Entry<String, io.netty.channel.Channel> entry : entrySet) {
								String clientId = entry.getKey();
								io.netty.channel.Channel ch = entry.getValue();
								ch.writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(res)));
							}
						}
					}
				}

			};
			channel.basicConsume(rabbitModel.getQueue(), rabbitModel.isAutoAck(), consume);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

}
